package org.parklab.lab.web.server;

import org.parklab.lab.web.server.PMF;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collections;
import javax.cache.Cache;
import javax.cache.CacheException;
import javax.cache.CacheManager;

import java.util.ArrayList;
import java.util.Date;
import java.util.Hashtable;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.jdo.PersistenceManager;

import org.parklab.lab.web.client.ChatterDB;
import org.parklab.lab.web.client.RTCDB;
import org.parklab.lab.web.client.RTCService;
import com.google.gwt.user.server.rpc.RemoteServiceServlet;

public class RTCServiceImpl extends RemoteServiceServlet implements RTCService {
	private static final Logger log = Logger.getLogger(RTCServiceImpl.class.getName());
	private static final long serialVersionUID = 2923971149841638945L;

	Cache chatInfoCache;
	Cache RTCache;
	
	private Hashtable<String, Hashtable<String, SynchronousQueue<RTCDB>>> queueHash;
	
	@Override
	public RTCDB getRTC(String userKey, String chatKey) {
		lineDebug("get:1");
		PrepareSynchronousQueue(userKey, chatKey);
		lineDebug("get:2");
		SynchronousQueue<RTCDB> q = queueHash.get(chatKey).get(userKey);
		lineDebug("get:3");
		try {
			lineDebug("get:4\t"+ q.hashCode());
			RTCDB r = q.poll(5, TimeUnit.SECONDS);
			if (r!=null) {
				lineDebug("get:5\t" + r.toString());
			} else {
				lineDebug("get:8");
			}
			return r;
		} catch (InterruptedException e) {
			lineDebug("get:6");
			log.warning("RTC Retrieve Fail: (userkey:" + userKey + ",chatKey:" + chatKey + ")");
			lineDebug("get:7");
			return null;
		}
	}

	private void lineDebug(String msg) {
		log.severe((new Date()).toString() + "\t" + msg);
	}

	@Override
	public RTCDB sendRTC(RTCDB myMessage) {
		lineDebug("send:1");
		String userKey = myMessage.getUserKey();
		lineDebug("send:2");
		String chatKey = myMessage.getChatKey();
		lineDebug("send:3");
		PrepareSynchronousQueue(userKey, chatKey);
		lineDebug("send:4");
		ArrayList<String> idiotList = new ArrayList<String>();
		lineDebug("send:5");
		for (Entry<String, SynchronousQueue<RTCDB>> q : queueHash.get(chatKey).entrySet()) {
			lineDebug("send:6");
			try {
				lineDebug("send:7\t" + q.hashCode());
				if (!q.getValue().offer(myMessage, 5, TimeUnit.SECONDS)) {
					lineDebug("send:8");
					idiotList.add(q.getKey());
					lineDebug("send:9");
				}
				lineDebug("send:10");
			} catch (InterruptedException e1) {
				lineDebug("send:11");
				log.warning("RTC Insert Fail: " + myMessage.toString() + ", to " + q.getKey());
				lineDebug("send:12");
			}
			lineDebug("send:13");
		}
		lineDebug("send:14");
		
		for (String aKey : idiotList) {
			lineDebug("send:15");
			synchronized (queueHash) {
				queueHash.get(chatKey).remove(aKey);
			}
			lineDebug("send:16");
		}
		lineDebug("send:17");
	
		if (idiotList.isEmpty()) {
			lineDebug("send:18");
			return myMessage;
		} else {
			lineDebug("send:19");
			myMessage.setText(myMessage.getText() + idiotList.get(0));
			return myMessage;
		}
	}

	
//	private void SaveChatUser(String chatKey, String userKey) {
//		renewCache(chatKey, userKey);
//	}
	
	private void renewCache(String chatKey, String usreKey) {
		if (chatInfoCache == null) {
			chatInfoCache = CacheManager.getInstance().getCache("chatInfo_" + chatKey);
			
			try {
				chatInfoCache = CacheManager.getInstance().getCacheFactory().createCache(Collections.emptyMap());
	        } catch (CacheException e) {
	            chatInfoCache = null;
	            logException(Level.SEVERE, e);
	        }
		}
	}

	private void logException(Level logLevel, Exception e) {
		StringWriter sw = new StringWriter();
		e.printStackTrace(new PrintWriter(sw));
		String stacktrace = sw.toString();
		log.log(logLevel, stacktrace);
	}

	@SuppressWarnings("unchecked")
	private void PrepareSynchronousQueue(String userKey, String chatKey) {
		
		synchronized (queueHash) {
			if (queueHash == null) {
				try {
		            chatInfoCache = CacheManager.getInstance().getCacheFactory().createCache(Collections.emptyMap());
		            if (chatInfoCache.containsKey("queueHash")) {
		            	queueHash = (Hashtable<String, Hashtable<String, SynchronousQueue<RTCDB>>>) chatInfoCache.get("queueHash");
		            } else {
		            	queueHash = new Hashtable<String, Hashtable<String,SynchronousQueue<RTCDB>>>();
		            }
		        } catch (CacheException e) {
		            
		        }
		        
		        chatInfoCache.put("queueHash", queueHash);
			} else if (!queueHash.containsKey(chatKey)) {
				SynchronousQueue<RTCDB> newQueue = new SynchronousQueue<RTCDB>(true);
				Hashtable<String, SynchronousQueue<RTCDB>> ht = new Hashtable<String, SynchronousQueue<RTCDB>>();
				ht.put(userKey, newQueue);
				queueHash.put(chatKey, ht);	
			} else if (!queueHash.get(chatKey).containsKey(userKey)) {
				SynchronousQueue<RTCDB> newQueue = new SynchronousQueue<RTCDB>(true);
				Hashtable<String, SynchronousQueue<RTCDB>> ht = queueHash.get(chatKey);
				ht.put(userKey, newQueue);
				queueHash.put(chatKey, ht);	
			}
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public List<ChatterDB> registerUser(String userKey, String chatKey) {
		ChatterDB aNewChatter = new ChatterDB(userKey, chatKey);
		
		PersistenceManager pm = PMF.get().getPersistenceManager();

        try {
            pm.makePersistent(aNewChatter);
        } catch (Exception e) {
        	
        }
        
        String query = "select from " + ChatterDB.class.getName();
        List<ChatterDB> chatters = (List<ChatterDB>) pm.newQuery(query).execute();
        pm.close();
        
        return chatters;
	}
	
	

}
